package spring.cloud.tasks.tasks_manager.tasks_manager.service;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.TreeCache;
import org.apache.curator.framework.recipes.cache.TreeCacheListener;
import org.apache.curator.utils.CloseableExecutorService;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;

@Slf4j
@Setter
@Getter
public class TreeCacheService {

    private String nameSpace;
    private CuratorFramework curatorFramework;
    private TreeCacheToTreeCacheListenerListMap treeCacheToTreeCacheListenerListMap;
    private ExecutorService executorService;
    private AtomicBoolean shutdown = new AtomicBoolean(true);

    public TreeCacheService(String nameSpace, CuratorFramework curatorFramework) {
        this.nameSpace = nameSpace;
        this.curatorFramework = curatorFramework;
        this.treeCacheToTreeCacheListenerListMap = new TreeCacheToTreeCacheListenerListMap();
    }

    public void start() {
        synchronized (shutdown) {
            if (shutdown.compareAndSet(true, false)) {
                shutdown0();
                executorService = Executors.newSingleThreadExecutor(new TreeCacheThreadFactory("NamespaceSharding-" + nameSpace));
            } else {
                log.warn("{}-ShardingTreeCacheService has already started, unnecessary to start", nameSpace);
            }
        }
    }


    public void shutdown() {
        synchronized (shutdown) {
            if (shutdown.compareAndSet(false, true)) {
                shutdown0();
            } else {
                log.warn("{}-ShardingTreeCacheService has already shutdown, unnecessary to shutdown", nameSpace);
            }
        }
    }

    private void shutdown0() {
        treeCacheToTreeCacheListenerListMap.shutdown();
        if (executorService != null) {
            executorService.shutdownNow();
        }
    }

    //
    public void addTreeCacheIfAbsent(String path, int depth) throws Exception {
        synchronized (shutdown) {
            if (shutdown.get()) {
                throw new Exception("ShardingTreeCacheService has been shutdown");
            }
            String fullPath = nameSpace + path;
            if (!treeCacheToTreeCacheListenerListMap.containsTreeCache(path, depth)) {
                TreeCache treeCache = TreeCache.newBuilder(curatorFramework, path).setMaxDepth(depth).setExecutor(new CloseableExecutorService(executorService, false)).build();
                try {
                    treeCache.start();
                } catch (Exception e) {
                    treeCache.close();
                    throw e;
                }
                TreeCache treeCacheOld = treeCacheToTreeCacheListenerListMap.addTreeCacheIfAbsent(path, depth, treeCache);
                if (treeCacheOld != null) {
                    treeCache.close();
                } else {
                    log.info("create TreeCache, full path is {}, depth is {}", fullPath, depth);
                }
            }
        }
    }

    public void addTreeCacheListenerIfAbsent(String path, int depth, TreeCacheListener treeCacheListener) throws Exception {
        synchronized (shutdown) {
            if (shutdown.get()) {
                throw new Exception("ShardingTreeCacheService has been shutdown");
            }
            String fullPath = nameSpace + path;
            TreeCacheListener treeCacheListenerOld = treeCacheToTreeCacheListenerListMap.addTreeCacheListenerIfAbsent(path, depth, treeCacheListener);
            if (treeCacheListenerOld == null) {
                log.info("add {}, full path is {}, depth is {}", treeCacheListener.getClass().getSimpleName(), fullPath, depth);
            }
        }
    }

    public void removeTreeCache(String path, int depth) throws Exception {
        synchronized (shutdown) {
            if (shutdown.get()) {
                throw new Exception("ShardingTreeCacheService has been shutdown");
            }

            treeCacheToTreeCacheListenerListMap.removeTreeCache(path, depth);
        }
    }


}
